package com.cnlive.logpare.app

import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.types.StringType
import com.cnlive.logpare.comm.CommRun
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.Row
import com.cnlive.logpare.util.IPSeekerExt
import scala.collection.mutable.HashMap
import com.cnlive.logpare.util.ParseLogUtil
import scala.collection.mutable.HashSet
import org.apache.spark.sql.SparkSession
import java.util.Date
import com.cnlive.logpare.comm.CommNginxRun
import com.cnlive.logpare.util.Constant
import org.apache.spark.sql.SQLContext

/**
 * Row类型分类 1.正常数据 2.直接被过滤条件过滤的数据 3.用 ""分割后，长度不对的数据
 *  4.appid不能分割的数据 5.url中用空格分割长度即不是3个也不是4个的一般都是4个少数为3个  6.url不带参数的
 */
object NewParseAppLogToHive extends CommRun with CommNginxRun{
    val logSign="app.js"
    val defaultAppName="NewParseAppLogToHive"
    def main(args: Array[String]): Unit = {
       val logSigns="/app.js?sver="
       val env=getEnv(args)
       val date=getDate(args)
       val files=getFilePath(env, logSign,date)
       val qqwryPath=getQqwryPath(env)//解析ip的文件地址
       val appname=getAppName(defaultAppName,date)
       val conf = new SparkConf().setAppName(appname)
       val installTime=tableNinxDateFormat.format(new Date) 
       if(env==1)conf.setMaster("local")
       val sc = new SparkContext(conf)
       val tabData= sc.textFile(files).mapPartitions(x=>{
         var result = List[Row]()
         val iPSeekerExt=getIpSeekerExt(env,qqwryPath)
         val sessionSet=new HashSet[String]
         while (x.hasNext) {
           val lineText=x.next()
           try{
               result= result.::(parLineNginxLogText(iPSeekerExt,lineText,sessionSet,parAppLogUrlandtypeToMap,mkApplogRow,filter,installTime))
           }catch {
               case t: Throwable =>println("正行抛出异常"+lineText); t.printStackTrace() 
           }
         }
         result.iterator
       },true)
       .filter(x=>x.getInt(0)==1).repartition(1)
        val querySql="""
            select count(distinct(uid)) uv, version,
             plat,sum(isnewuser)as au,
             sum(isstart) as pv from appLogTempTab
             where year='2017'
              and moth='06' and day in ('22','23') and spid='3'
              and appid='3_it8a4s3k14' 
               group by  version,plat 
              order by pv desc 
              """
        val warehouseLocation= "spark-warehouse"
        val spark=getSparkSql(env,conf)
        import spark.implicits._
        val appLogtab=spark.createDataFrame(tabData, appLogSchema)
           appLogtab.registerTempTable("appLogTempTab")
        val installSql="""
             insert into TABLE  appofflog partition(year,moth,day,spid) 
             select nginxtime,appid,uid,ip,version,city,location,operator,
             plat,eventid,isnewuser,isstart,installtime,year,moth,day,spid from  appLogTempTab
              """
         spark.sql(installSql)
    }
   val filter=(lineText:String)=>{
      if(lineText.indexOf(logSign) != -1) true else false
   }
   //解析url中的数据到map中
   val  parAppLogUrlandtypeToMap=(urlandtype:List[String],mapRequest:HashMap[String, String],sessionSet:HashSet[String])=>{
       val urlandper=urlandtype(0).split(" ")
       val d=urlandper match {
              case Array(_,_,z,_*) =>{  
                  ParseLogUtil.parseUrlToMap(z, mapRequest)//解析url中的参数
                  parIsNewUser(mapRequest,sessionSet)
              }
              case _=>mapRequest.put(Constant.errCodeKey, Constant.errCodeValue_001)
       }
   }
   //生成appLog的一行数据
   val mkApplogRow=(mapRequest:HashMap[String, String])=>{
            mapRequest.get(Constant.errCodeKey)  match {
                case Some(codeValue) =>if(codeValue.equals(Constant.errCodeValue_001)) Row(5)else Row(6)
                case _=>{
                  val appid=mapRequest.getOrElse("appid", "null")
                    val spidandappid=appid.split("_")
                    if(spidandappid.length!=2){
                       Row(4)
                    }else{
                       val nginxTime=tableNinxDateFormat.format(nginxDateFormat.parse(mapRequest.getOrElse("nginxtime", "19700101000000")))
                       val uid=mapRequest.getOrElse("uid", "0") 
                       val ip=mapRequest.getOrElse("ip", "127.0.0.1")
                       val version=mapRequest.getOrElse("version", "0_ ").split("_")  match {
                         case Array(_,y,_*) =>y
                         case Array(x) =>x
                       }
                       val city=mapRequest.getOrElse("city", "unknown")//默认值
                       val location=mapRequest.getOrElse("location", "unknown")//默认值
                       val operator=mapRequest.getOrElse("operator", "99").toInt
                       val plat=mapRequest.getOrElse("plat", " ")  match {
                           case "i" =>103
                           case "a" =>101
                           case _=>199
                       }
                       val eventid=mapRequest.getOrElse("eventid", "unknown")//默认值
                       val isnewuser=mapRequest.getOrElse("isnewuser", "0").toInt//默认值
                       val installtime=mapRequest.getOrElse("installtime", "19700101000000")//默认值
                       val isstart=eventid match {
                          case "1300" =>1
                          case _ =>0 
                       }
                       Row(1,nginxTime.toLong,appid,uid,ip,version,city,location,operator,plat,eventid,isnewuser,isstart,installtime.toLong,nginxTime.substring(0, 4),nginxTime.substring(4, 6),nginxTime.substring(6, 8),spidandappid(0))
                    }
                }
           }
    }
    //解析，看是不是一个新的用户
    def parIsNewUser(mapRequest:HashMap[String, String],sessionSet:HashSet[String]){
       if(mapRequest.contains("uid")&&mapRequest.contains("tm")){
          val uid=mapRequest.getOrElse("uid", "")
          val tm=mapRequest.getOrElse("tm", "")
          val uidAndtm=uid.split("_")
          if(uidAndtm.length==2){
            if(uidAndtm(1).length()==13&&uidAndtm(1).matches("^[0-9]*$")){
              if(uidAndtm(1).equals(tm)){
                 if(sessionSet.contains(uid)){
                    mapRequest.put("isnewuser", "0")  
                 }else{
                    sessionSet.add(uid)
                    mapRequest.put("isnewuser", "1")  
                 }
               }else{
                  mapRequest.put("isnewuser", "0")
               }
            }else if(uidAndtm(0).length()==13&&uidAndtm(0).matches("^[0-9]*$")){
              if(uidAndtm(0).equals(tm)){
                 if(sessionSet.contains(uid)){
                    mapRequest.put("isnewuser", "0")   
                 }else{
                    sessionSet.add(uid)
                    mapRequest.put("isnewuser", "1")   
                 }
               }else{
                  mapRequest.put("isnewuser", "0")
               }
            }
          }else{
             mapRequest.put("isnewuser", "0")
          }
       }else{
         mapRequest.put("isnewuser", "0")
       }
    }
    //appLog表结构
    val appLogSchema = StructType(Seq(
        StructField("datatype", IntegerType, nullable = true),
        StructField("nginxtime", LongType, nullable = true),
        StructField("appid", StringType, nullable = true),
        StructField("uid", StringType, nullable = true),
        StructField("ip", StringType, nullable = true),
        StructField("version", StringType, nullable = true),
        StructField("city", StringType, nullable = true),
        StructField("location", StringType, nullable = false),
        StructField("operator", IntegerType, nullable = false),
        StructField("plat", IntegerType, nullable = false),
        StructField("eventid", StringType, nullable = false),
        StructField("isnewuser", IntegerType, nullable = false),
        StructField("isstart", IntegerType, nullable = false),
        StructField("installTime", LongType, nullable = true),
        StructField("year", StringType, nullable = true),
        StructField("moth", StringType, nullable = true),
        StructField("day", StringType, nullable = true),
        StructField("spid", StringType, nullable = true)
      ))
}